package com.cloudwise.monitoring_system.util;

import com.cloudwise.monitoring_system.entity.MonitorEvent;
import com.cloudwise.monitoring_system.entity.MonitorUnit;
import com.cloudwise.monitoring_system.entity.MonitorValue;
import com.cloudwise.monitoring_system.service.clickhouseService.impl.CkMonitorEventServiceImpl;
import com.cloudwise.monitoring_system.service.clickhouseService.impl.CkMonitorValueServiceImpl;
import com.cloudwise.monitoring_system.service.mysqlService.impl.MonitorEventServiceImpl;
import com.cloudwise.monitoring_system.service.mysqlService.impl.MonitorUnitServiceImpl;
import com.cloudwise.monitoring_system.service.mysqlService.impl.MonitorValueServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Controller;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @author Reisen
 * @description: kafka消费者
 * @date 2021/5/18 10:26
 */
@Controller
@Slf4j
public class KafkaConsumer {
        @Resource
        private MonitorValueServiceImpl monitorValueService;
        @Resource
        private MonitorUnitServiceImpl monitorUnitService;
        @Resource
        private MonitorEventServiceImpl monitorEventService;
        @Resource
        private CkMonitorValueServiceImpl ckMonitorValueService;
        @Resource
        private CkMonitorEventServiceImpl ckMonitorEventService;
    /**
     * 低频数据写入mysql
     * @param record
     */
    @KafkaListener(topics = {"test1"})
        public void consumer(ConsumerRecord<?, ?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            if (kafkaMessage.isPresent()) {
                Object message = kafkaMessage.get();
                //System.out.println(message);
                String result = message.toString();
                if (result!=null){
                    //将日志文件按空格拆分成数组
                    String[] s = result.split(" ");
                    MonitorValue monitorValue = new MonitorValue();
                    MonitorUnit monitorUnit = new MonitorUnit();
                    monitorUnit.setUnitName(s[3]);
                    MonitorUnit ResultUint = monitorUnitService.queryByName(monitorUnit);

                    if (ResultUint!=null){
                        Integer unitId = ResultUint.getUnitId();
                        monitorValue.setCreateTime(s[0]+" "+s[1]);
                        monitorValue.setUnitId(unitId);
                        monitorValue.setMonitorValue(Double.parseDouble(s[4]));
                        if (monitorValue.getMonitorValue()>ResultUint.getMax() || monitorValue.getMonitorValue()<ResultUint.getMin()){
                            MonitorEvent monitorEvent = new MonitorEvent();
                            monitorEvent.setUnitId(monitorValue.getUnitId());
                            monitorEvent.setCreateTime(monitorValue.getCreateTime());
                            monitorEvent.setMonitorValue(monitorValue.getMonitorValue());
                            monitorEvent.setUnitId(monitorValue.getUnitId());
                            log.info("mysql数据库新增一条告警信息");
                            monitorEventService.insert(monitorEvent);
                        }
                            log.info("mysql数据库新增一条监控项数据"+monitorValue);
                            monitorValueService.insert(monitorValue);
                    }
                }
            }
        }
    /**
     * 低频数据写入mysql
     * @param record
     */
    @KafkaListener(topics = {"test2"})
    public void consumer2(ConsumerRecord<?, ?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            log.info("开始消费数据");
            Object message = kafkaMessage.get();
            //System.out.println(message);
            String result = message.toString();
            if (result!=null){
                //将日志文件按空格拆分成数组
                String[] s = result.split(" ");
                MonitorValue monitorValue = new MonitorValue();
                MonitorUnit monitorUnit = new MonitorUnit();
                monitorUnit.setUnitName(s[3]);
                MonitorUnit ResultUint = monitorUnitService.queryByName(monitorUnit);
                if (ResultUint!=null){
                    Integer unitId = ResultUint.getUnitId();
                    monitorValue.setCreateTime(s[0]+" "+s[1]);
                    monitorValue.setUnitId(unitId);
                    monitorValue.setMonitorValue(Double.parseDouble(s[4]));
                    if (monitorValue.getMonitorValue()>ResultUint.getMax() || monitorValue.getMonitorValue()<ResultUint.getMin()){
                        MonitorEvent monitorEvent = new MonitorEvent();
                        monitorEvent.setUnitId(monitorValue.getUnitId());
                        monitorEvent.setCreateTime(monitorValue.getCreateTime());
                        monitorEvent.setMonitorValue(monitorValue.getMonitorValue());
                        monitorEvent.setUnitId(monitorValue.getUnitId());
                        log.info("clickhouse新增一条告警信息");
                        ckMonitorEventService.insert(monitorEvent);


                    }else {
                        log.info("clickhouse新增一条监控项数据"+monitorValue);
                        ckMonitorValueService.insert(monitorValue);
                    }


                }
            }
        }
    }

}
